package com.maple.retry.retrytask.manager;

import com.maple.retry.retrytask.RetryTaskExecutor;
import com.maple.retry.retrytask.RetryTaskResult;
import com.maple.retry.retrytask.RetryTaskRunner;
import com.maple.retry.retrytask.ScheduleStrategy;
import com.maple.retry.retrytask.config.RetryTaskConfig;
import com.maple.retry.retrytask.enums.RetryTaskStatusEnum;
import com.maple.retry.retrytask.model.RetryTaskContext;
import com.maple.retry.retrytask.model.RetryTaskInfo;
import com.maple.retry.retrytask.util.RetryTaskDigestLogUtil;
import com.maple.retry.service.RetryTaskService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallbackWithoutResult;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author: maple
 * @version: RetryTaskManagerImpl.java, v 0.1 2020年12月13日 19:58 maple Exp $
 */
@Slf4j
@Component
public class RetryTaskManagerImpl implements RetryTaskManager {

    /**
     * 补偿任务业务类型和配置映射关系
     */
    private Map<String, RetryTaskConfig> configs = new ConcurrentHashMap<>();

    /**
     * 补偿任务服务接口
     */
    @Autowired
    private RetryTaskService retryTaskService;

    /**
     * 事务模板
     */
    @Autowired
    private TransactionTemplate transactionTemplate;

    /**
     * 补偿任务执行线程池
     */
    @Autowired
    private ThreadPoolTaskExecutor executor;

    @Override
    public void register(String bizType, RetryTaskConfig config) {

        long start = System.currentTimeMillis();

        RetryTaskResult result = null;

        try {
            // 1、参数检查
            // 参数非空校验
            Assert.hasText(bizType, "bizType is null");
            Assert.notNull(config, "compensate task config is null");

            // 业务校验
            config.validation();
            Assert.isTrue(!configs.containsKey(bizType), "bizType:" + bizType + " has bean registered");

            configs.put(bizType, config);
            result = RetryTaskResult.valueOfSuccess((RetryTaskInfo) null);
        } catch (Throwable e) {

            //注册过程中出现异常,抛运行时异常，服务启动中断
            log.error("register failed, bizType=", bizType);
            result = RetryTaskResult.valueOfError(e);
            throw new RuntimeException(e);

        } finally {

            long elapseTime = System.currentTimeMillis() - start;
            log.info(RetryTaskDigestLogUtil.buildLifecycleDigestLog("create", elapseTime, result.isSuccess(), null, bizType));
        }
    }

    @Override
    public RetryTaskResult create(final String bizId, final String bizType, final RetryTaskContext context) {

        long start = System.currentTimeMillis();

        RetryTaskResult result = null;

        try {

            result = transactionTemplate.execute(transactionStatus -> {

                // 1、参数检查
                checkParams(bizId, bizType, context);

                // 2、唯一性校验，查询历史表
                RetryTaskInfo historyTaskInfo = retryTaskService.loadArchivedTaskByBizInfo(bizId, bizType);
                if (null != historyTaskInfo) {
                    throw new RuntimeException("task has existed, bizId=" + bizId + ", bitType=" + bizType);
                }

                // 3、模型组装
                RetryTaskInfo taskInfo = assembleRetryTaskInfo(bizId, bizType, context.toString());

                // 4、新增重试表
                retryTaskService.create(taskInfo, configs.get(bizType).getDelayInMills());

                return RetryTaskResult.valueOfSuccess(taskInfo);
            });

        } catch (Throwable e) {
            log.error("create compensate task failed, bizId=", bizId, ", bizType=", bizType, e);
            result = RetryTaskResult.valueOfError(e);
        } finally {

            long elapseTime = System.currentTimeMillis() - start;
            log.info(RetryTaskDigestLogUtil.buildLifecycleDigestLog("create", elapseTime, result.isSuccess(), bizId, bizType));
        }

        return result;
    }

    @Override
    public RetryTaskResult abort(String bizId, String bizType) {
        return null;
    }

    @Override
    public RetryTaskResult process(String bizType) {

        List<RetryTaskInfo> taskInfos = new ArrayList<>();

        try {

            //1. 消息业务校验
            bizCheck(bizType);

            //2. 任务前置处理
            preprocess(taskInfos, bizType);

            //3. 异步调度任务执行,包括结果处理
            asyncProcess(taskInfos, bizType);

            return RetryTaskResult.valueOfSuccess(taskInfos);

        } catch (Throwable e) {
            log.error("retry task process failed, bizType=", StringUtils.defaultIfBlank(bizType, ""));
            return RetryTaskResult.valueOfError(e);
        }
    }

    /**
     * 执行补偿任务
     *
     * @param taskInfo 补偿任务
     * @param executor 补偿任务执行器
     * @param strategy 任务结果决策
     * @return 执行结果
     */
    public RetryTaskResult innerProcess(RetryTaskInfo taskInfo, RetryTaskExecutor executor, ScheduleStrategy strategy) {

        List<RetryTaskInfo> taskInfos = new ArrayList<>();
        RetryTaskResult result = null;
        long start = System.currentTimeMillis();

        try {

            RetryTaskContext retryTaskContext = RetryTaskContext.valueOf(taskInfo.getBizContext());

            //0. 恢复上下文
            //initContext(retryTaskContext);

            // Update task from INIT to PROCESSING
            if (!updateInitTaskToProcessingStatus(taskInfo)) {
                log.info("Skipping update task for=" + taskInfo.getId());
                // return as success to same behave as per retry and fail
                return result = RetryTaskResult.valueOfSuccess();
            }

            //1. 任务执行
            RetryTaskResult executeResult = executor.execute(retryTaskContext);

            //2. 把原始任务回填到结果中
            taskInfos.add(taskInfo);
            executeResult.setRetryTaskInfos(taskInfos);

            //3. 执行结果处理
            result = finish(executeResult, strategy);

        } catch (Throwable e) {

            log.error("compensate task inner process failed");
            result = RetryTaskResult.valueOfError(e);

            // Revert status back to init
            revertToInitStatus(taskInfo);
        } finally {

            long elapseTime = System.currentTimeMillis() - start;

            log.info(RetryTaskDigestLogUtil.buildProcessDigestLog(elapseTime, result.isSuccess(), taskInfo.getBizId(), taskInfo.getBizType(),
                    taskInfo.getStatus().getCode(), strategy.isForever(),
                    String.valueOf(taskInfo.getRetryNo())));
        }

        return result;
    }


    private RetryTaskInfo assembleRetryTaskInfo(String bizId, String bizType,
                                                String bizContext) {

        RetryTaskInfo taskInfo = new RetryTaskInfo();

        taskInfo.setBizId(bizId);
        taskInfo.setBizType(bizType);
        taskInfo.setBizContext(bizContext);

        return taskInfo;
    }

    private void checkParams(String bizId, String bizType, RetryTaskContext retryTaskContext) {
        int bizIdLength = 65;
        int bizTypeLength = 65;
        int bizContextLength = 20000;

        Assert.notNull(retryTaskContext, "unknow biz context, is null.");
        String context = retryTaskContext.toString();

        Assert.hasText(bizId, "bizId is empty");
        Assert.hasText(bizType, "bizType is blank");
        Assert.hasText(context, "bizContext is blank");

        Assert.isTrue(bizId.length() < 65, "bizId expected length is[" + bizIdLength
                + "], current length is [" + bizId.length() + "]");
        Assert.isTrue(bizType.length() < 65, "bizType expected length is[" + bizTypeLength
                + "], current length is [" + bizType.length() + "]");
        Assert
                .isTrue(context.length() < 20000, "bizContext expected length is[" + bizContextLength
                        + "], current length is [" + context.length() + "]");

        Assert.notNull(configs.get(bizType), "unknown bizType, bizType=" + bizType);
    }

    private void bizCheck(String bizType) {

        Assert.hasText(bizType, "bizType is blank");
        Assert.notNull(configs.get(bizType), "unknown bizType, bizType=" + bizType);
        Assert.notNull(executor, "system error: thread pool initial failed");
    }

    /**
     * 任务执行前置处理流程
     *
     * @param taskInfos 补偿任务列表, 作为引用传入
     * @param bizType   业务类型
     */
    private void preprocess(List<RetryTaskInfo> taskInfos, String bizType) {
        // 获取补偿任务列表
        taskInfos.addAll(retryTaskService.loadScheduleTasksByStatus(bizType, Integer.valueOf(configs.get(bizType).getLoadSize()), RetryTaskStatusEnum.INIT));
    }

    /**
     * 异步执行补偿任务
     *
     * @param taskInfos 补偿任务列表
     * @param bizType   补偿任务业务类型
     */
    private void asyncProcess(List<RetryTaskInfo> taskInfos, String bizType) {

        for (RetryTaskInfo taskInfo : taskInfos) {

            RetryTaskRunner runner = null;

            try {

                runner = new RetryTaskRunner(taskInfo, configs.get(bizType).getExecutor(), this, configs.get(bizType).getScheduleStrategy());
                executor.submit(runner);

            } catch (Throwable e) {
                log.error("{}", e);
            }
        }
    }

    private boolean updateInitTaskToProcessingStatus(RetryTaskInfo taskInfo) {

        return transactionTemplate.execute(transactionStatus -> {

            RetryTaskInfo dbTaskInfo = retryTaskService.lockByTaskId(taskInfo.getId());

            // Task finished process or still in processing by others
            if (dbTaskInfo == null || RetryTaskStatusEnum.INIT != dbTaskInfo.getStatus()) {
                return false;
            }

            retryTaskService.updateStatusByTaskId(dbTaskInfo, RetryTaskStatusEnum.PROCESSING);

            return true;
        });
    }

    private RetryTaskResult finish(RetryTaskResult executeResult, ScheduleStrategy scheduleStrategy) {

        // 补偿任务执行成功
        if (executeResult.isSuccess()) {
            handleSuccessTask(executeResult.fetchSingleRetryTask());
            return RetryTaskResult.valueOfSuccess(executeResult.getRetryTaskInfos());
        }

        if (CollectionUtils.isEmpty(executeResult.getRetryTaskInfos())) {
            log.warn("no compensate task found");
            return RetryTaskResult.valueOfError(new Exception("no compensate task found"));
        }

        int retryNo = executeResult.fetchSingleRetryTask().getRetryNo();

        // 补偿任务需要重试
        if (retry(scheduleStrategy, retryNo)) {
            handleRetryTask(executeResult.fetchSingleRetryTask(), scheduleStrategy.nextDelayTime(retryNo));
            return RetryTaskResult.valueOfSuccess(executeResult.getRetryTaskInfos());
        }

        // 补偿任务失败且超过重试次数
        handleFailedTask(executeResult.fetchSingleRetryTask());
        return RetryTaskResult.valueOfSuccess(executeResult.getRetryTaskInfos());
    }

    /**
     * 对执行成功的任务处理
     *
     * @param retryTaskInfo 原始补偿任务
     */
    private void handleSuccessTask(final RetryTaskInfo retryTaskInfo) {

        transactionTemplate.execute(new TransactionCallbackWithoutResult() {

            @Override
            public void doInTransactionWithoutResult(TransactionStatus status) {

                //1. 锁原始任务
                RetryTaskInfo dbTaskInfo = retryTaskService.lockByTaskId(retryTaskInfo.getId());

                //防止并发场景下,任务由另一个线程执行且事务已提交造成原始任务已归档
                if (dbTaskInfo == null) {
                    return;
                }

                //2. 删除原始补偿任务
                retryTaskService.drop(retryTaskInfo.getId());

                //3. 归档补偿任务记录,状态为success
                archiveCompensateTask(dbTaskInfo, RetryTaskStatusEnum.SUCCESS);
            }
        });
    }

    /**
     * 对需要重试的任务处理
     *
     * @param retryTaskInfo 原补偿任务
     * @param delay         下次执行延迟时间
     */
    private void handleRetryTask(final RetryTaskInfo retryTaskInfo, final long delay) {

        transactionTemplate.execute(new TransactionCallbackWithoutResult() {

            @Override
            protected void doInTransactionWithoutResult(TransactionStatus status) {

                //1. 锁原始任务
                RetryTaskInfo dbTaskInfo = retryTaskService.lockByTaskId(retryTaskInfo.getId());

                // 防止并发场景下,任务由另一个线程执行且事务已提交造成原始任务已归档
                if (dbTaskInfo == null) {
                    return;
                }

                //2. 更新原始任务
                retryTaskService.updateRetryAndStatusByTaskId(dbTaskInfo, RetryTaskStatusEnum.INIT, delay);

                // 因为service层更新的数据不会带出来,需要手动retryNo做累加操作
                retryTaskInfo.increaseRetryNo();
            }
        });
    }

    /**
     * 对执行失败且超过最大重试次数的任务处理
     *
     * @param retryTaskInfo 补偿任务
     */
    private void handleFailedTask(final RetryTaskInfo retryTaskInfo) {

        transactionTemplate.execute(new TransactionCallbackWithoutResult() {

            @Override
            protected void doInTransactionWithoutResult(TransactionStatus status) {

                //1. 锁原始任务
                RetryTaskInfo dbTaskInfo = retryTaskService.lockByTaskId(retryTaskInfo.getId());

                //防止并发场景下,任务由另一个线程执行且事务已提交造成原始任务已归档
                if (dbTaskInfo == null) {
                    return;
                }

                //2. 删除原始补偿任务
                retryTaskService.drop(retryTaskInfo.getId());

                //3. 归档补偿任务记录,状态为failed
                archiveCompensateTask(retryTaskInfo, RetryTaskStatusEnum.FAILED);
            }
        });
    }

    /**
     * Revert back to INIT status
     *
     * @param taskInfo CompensateTaskInfo
     */
    private void revertToInitStatus(RetryTaskInfo taskInfo) {
        transactionTemplate.execute(new TransactionCallbackWithoutResult() {
            @Override
            protected void doInTransactionWithoutResult(TransactionStatus transactionStatus) {
                RetryTaskInfo dbTaskInfo = retryTaskService.lockByTaskId(taskInfo.getId());

                // Could be processed already by others
                if (dbTaskInfo == null || RetryTaskStatusEnum.INIT == dbTaskInfo.getStatus()) {
                    return;
                }

                retryTaskService.updateStatusByTaskId(dbTaskInfo, RetryTaskStatusEnum.INIT);
            }
        });
    }

    private boolean retry(ScheduleStrategy scheduleStrategy, int retryNo) {
        return scheduleStrategy.isForever() || retryNo < scheduleStrategy.getLongRetryIntervals().length;
    }

    /**
     * 归档补偿任务
     * <p>
     * 将任务领域模型状态置为"abort"
     * </p>
     *
     * @param taskInfo 补偿任务领域模型
     * @param status   补偿任务归档后的状态
     */
    private void archiveCompensateTask(RetryTaskInfo taskInfo, RetryTaskStatusEnum status) {

        taskInfo.setStatus(status);
        retryTaskService.archive(taskInfo);
    }

    /**
     * Setter method for property configs.
     *
     * @param configs value to be assigned to property configs
     */
    public void setConfigs(Map<String, RetryTaskConfig> configs) {
        this.configs = configs;
    }

    /**
     * Getter method for property configs.
     *
     * @return property value of configs
     */
    public Map<String, RetryTaskConfig> getConfigs() {
        return configs;
    }
}
